#include "pushgateway_client.h"
#include "common.h"
#include "utils.h"
#include "helper.h"
#include "scaler.h"
#include "agent.h"

#include <thread>
#include <map>
#include <string>
#include <vector>
#include <algorithm>
#include <chrono>
#include <curl/curl.h>
#include <unistd.h>
#include <malloc.h>
#include <iostream>
#include <fstream>
#include <sys/stat.h>
#include <sys/file.h>
#include <stdlib.h>
#include <stdio.h>
#include <stddef.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
#include <sys/types.h>

bool compareDoubleString(const std::string &first, const std::string &second)
{
    char first_c[first.length() + 1];
    char second_c[second.length() + 1];
    strcpy(first_c, first.c_str());
    strcpy(second_c, second.c_str());
    double first_d = atof(first_c);
    double second_d = atof(second_c);

    return first_d < second_d + EPS;
}

PushGatewayClient *PushGatewayClient::newInstance()
{
    if (PushGatewayClient::instance == nullptr)
    {
        PushGatewayClient::instance = new PushGatewayClient();
        PushGatewayClient::instance->pid = getpid();
    }
    return PushGatewayClient::instance;
}

void PushGatewayClient::send(std::string collect, long size)
{
    logger("[pushgateway] send start");
    PushGatewayClient *pushgateway_client = PushGatewayClient::newInstance();

    // 数据是否为空
    if (!pushgateway_client->isEmpty())
    {
        logger("[pushgateway] fetch content");
        std::string request = pushgateway_client->getData();

        // logger("[pushgateway] " + request);

        // 此方法需要升级 libcurl 7.40.0
        // CURL *curl = curl_easy_init();

        // curl_easy_setopt(curl, CURLOPT_UNIX_SOCKET_PATH, collect.c_str());
        // curl_easy_setopt(curl, CURLOPT_URL, "http://localhost/");
        // curl_easy_setopt(curl, CURLOPT_POST, 1);
        // curl_easy_setopt(curl, CURLOPT_POSTFIELDS, request.c_str());
        // curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 1);
        // curl_easy_setopt(curl, CURLOPT_TIMEOUT, 1);

        // curl_easy_perform(curl);
        // curl_easy_cleanup(curl);

        int sock;
        struct sockaddr_un server;

        sock = socket(AF_UNIX, SOCK_STREAM, 0);
        if (sock < 0)
        {
            logger("[pushgateway] open socket fail");
        }
        else
        {
            logger("[pushgateway] open socket success");
            server.sun_family = AF_UNIX;
            strcpy(server.sun_path, collect.c_str());

            if (connect(sock, (struct sockaddr *)&server, sizeof(struct sockaddr_un)) < 0)
            {
                close(sock);
                logger("[pushgateway] connect socket fail");
            }
            else
            {
                logger("[pushgateway] connect socket success");
                if (write(sock, request.c_str(), request.length()) < 0)
                {
                    logger("[pushgateway] write socket fail");
                }
                else
                {
                    logger("[pushgateway] write socket success");
                }
                close(sock);
            }
        }
    }
    else
    {
        logger("[pushgateway] content not found");
    }

    pushgateway_client->clean();
}

std::string PushGatewayClient::getData()
{
    return this->data;
}

void PushGatewayClient::setCounter(Scaler *sc)
{
    std::string uri = pushgateway_uri();
    std::string index = sc->getIndex();
    std::string scStr = sc->toString();
    // std::string res = "# TYPE " + index + " counter#URL:" + uri + "\n" + scStr + " 1\n";
    std::string res = "# URL:" + uri + "\n# TYPE " + index + " counter\n" + scStr + " 1\n";

    this->data += res;
}

void PushGatewayClient::setBucket(Scaler *sc, double seconds)
{
    std::string uri = pushgateway_uri();

    // 获取bucket范围
    std::string bucket_opt = MONITORPRO_G(pushbucket);
    std::vector<std::string> buckets = split(bucket_opt, ",");
    // buckets排序
    sort(buckets.begin(), buckets.end(), compareDoubleString);

    std::map<std::string, std::string> items = sc->getItems();
    std::string index = sc->getIndex();
    // std::string res = "# TYPE " + index + " histogram#URL:" + uri + "\n";
    std::string res = "# URL:" + uri + "\n# TYPE " + index + " histogram";

    Scaler *sc_bucket = nullptr;
    for (auto bucket : buckets)
    {
        std::string sc_bucket_index = index + "_bucket";
        sc_bucket = new Scaler();
        sc_bucket->setIndex(sc_bucket_index);
        sc_bucket->setItems(items);
        sc_bucket->addItem(ITEM_LE, bucket);
        std::string itemStr = sc_bucket->toString();
        std::string count = "0";
        if (!compareDoubleString(bucket, std::to_string(seconds)))
        {
            count = "1";
        }
        res += itemStr + " " + count + "\n";

        // 清理内存
        sc_bucket->clean();
        delete sc_bucket;
        sc_bucket = nullptr;
    }

    // 计算bucket时间累积总数
    std::string sc_bucket_sum_index = index + "_sum";
    Scaler *sc_bucket_sum = new Scaler;
    sc_bucket_sum->setIndex(sc_bucket_sum_index);
    sc_bucket_sum->setItems(items);
    std::string bucket_sum_key = sc_bucket_sum->toString();
    long seconds_i = seconds * pow(10, MONITORPRO_HISTOGRAM_DECIAML);
    double seconds_d = seconds_i * pow(0.1, MONITORPRO_HISTOGRAM_DECIAML);
    res += bucket_sum_key + " " + std::to_string(seconds_d) + "\n";

    sc_bucket_sum->clean();
    delete sc_bucket_sum;
    sc_bucket_sum = nullptr;

    // 计算bucket请求累积总数
    std::string sc_bucket_count_index = index + "_count";
    Scaler *sc_bucket_count = new Scaler;
    sc_bucket_count->setIndex(sc_bucket_count_index);
    sc_bucket_count->setItems(items);
    std::string bucket_count_key = sc_bucket_count->toString();
    res += bucket_count_key + " 1\n";

    sc_bucket_count->clean();
    delete sc_bucket_count;
    sc_bucket_count = nullptr;

    // 写入data
    this->data += res;

    buckets.clear();
    items.clear();
    std::vector<std::string>().swap(buckets);
    std::map<std::string, std::string>().swap(items);
    malloc_trim(0);
}

void PushGatewayClient::run()
{
    logger("[pushgateway] running");
    std::string collect = MONITORPRO_G(pushcollect);
    long size = MONITORPRO_G(pushcollectsize);
    long buf = MONITORPRO_G(pushbuf);
    long interval = MONITORPRO_G(pushflushinterval);
    long totalbuf = 1048576 * buf;
    long totalsize = 1048576 * size;

    if (collect == "")
    {
        logger("[pushgateway] monitorpro.pushcollect is empty");
        return;
    }

    // logger("[pushgateway] totalbuf: " + std::to_string(totalbuf));
    // logger("[pushgateway] totalsize: " + std::to_string(this->data.size()));

    if ((totalbuf <= 0 && this->isInterval()) || (totalbuf > 0 && ((long)this->data.size() >= totalbuf || (interval > 0 && this->isInterval()))))
    {
        std::thread th(send, collect, totalsize);
        th.detach();
    }
}

void PushGatewayClient::clean()
{
    this->data.clear();
    this->data = "";
}

bool PushGatewayClient::isEmpty()
{
    return this->data == "";
}

bool PushGatewayClient::isInterval()
{
    long interval = MONITORPRO_G(pushflushinterval);
    // 获取当前时间
    struct timeval tv;
    long now;
    gettimeofday(&tv, NULL);
    now = tv.tv_sec;

    // logger("[pushgateway] now: " + std::to_string(now));
    // logger("[pushgateway] lasttime: " + std::to_string(this->lasttime));
    if ((now - this->lasttime) >= interval)
    {
        this->lasttime = now;
        return true;
    }

    return false;
}